from pyspark.sql import SparkSession
import os

from cn.itcast.tag.bean.ESMeta import ruleToESMeta

"""
-------------------------------------------------
   Description :	TODO：演示通义灵码 AI编程助手实现性别标签
   SourceFile  :	GenderModel_AI
   Author      :	itcast team
-------------------------------------------------
"""

# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'

#1.创建SparkSession对象，以local[2]模式运行，SparkSQL的分区数为4
spark = SparkSession\
    .builder\
    .appName("GenderModel_AI")\
    .master("local[2]")\
    .config("spark.sql.shuffle.partitions","4")\
    .getOrCreate()

#2.读取MySQL数据源，数据库为tfec_tags，主机为up01，表为tbl_basic_tag，用户名为root，密码为123456，得到input_df对象
#url：url串
#dbtable 表名
#user 用户名
#password 密码
input_df = spark.read.format("jdbc")\
    .option("url","jdbc:mysql://up01:3306/tfec_tags")\
    .option("dbtable","tbl_basic_tag")\
    .option("user","root")\
    .option("password","123456")\
    .load()

input_df.show()
input_df.printSchema()

#3.取出上述MySQL数据中id为4的数据，得到four_df对象
four_df = input_df.filter(input_df.id == 4)
#根据four_df，取出rule字段的数据，并转换为字符串，得到rule_str
rule_str = four_df.select("rule").rdd.map(lambda x:x[0]).collect()[0]
#最后由rule_str字符串根据自定义函数ruleToESMeta，转换为ESMeta对象，命名为EsMeta
EsMeta = ruleToESMeta(rule_str)
#打印EsMeta
print(EsMeta)

#4.过滤上述MySQL数据中pid为4的数据，得到五级标签规则数据，选择其中的id和rule字段数据，结果命名为five_df
five_df = input_df.filter(input_df.pid == 4).select("id","rule")
#打印five_df
five_df.show()
five_df.printSchema()


#5.读取ElasticSearch中的数据，格式为es，resource为EsMeta中的esIndex属性，nodes为EsMeta中的esNodes属性，字段为EsMeta中的selectFields属性，最终返回es_df
es_df = spark.read.format("es")\
    .option("es.nodes",EsMeta.esNodes)\
    .option("es.resource",EsMeta.esIndex)\
    .option("es.read.field.include",EsMeta.selectFields)\
    .load()
#打印es_df
es_df.show()
es_df.printSchema()

#6.把es_df和five_df进行left join关联，条件是five_df的rule等于es_df的gender，选择es_df的id和five_df的id，类型转换为string，分别重命名为userId和tagsId，返回result_df
result_df = es_df.join(five_df,es_df.gender == five_df.rule,"left")\
    .select(es_df.id.cast("string").alias("userId"),five_df.id.cast("string").alias("tagsId"))

#7.把数据写出到ElasticSearch中，格式为es，resource为tag_result，nodes为EsMeta中的esNodes属性，并且指定id映射为数据的userId
result_df.write.format("es")\
    .option("es.nodes",EsMeta.esNodes)\
    .option("es.resource","tag_result")\
    .option("es.mapping.id","userId")\
    .save()

#8.关闭SparkSession对象
spark.stop()